package com.oracle.calllog.analyze.outputformat;

import java.io.IOException;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;

import com.oracle.calllog.analyze.model.ContactDimBean;
import com.oracle.calllog.analyze.model.CountDuration;
import com.oracle.calllog.analyze.util.ConnectionPool;
import com.oracle.calllog.analyze.util.Constants;
import com.oracle.calllog.analyze.util.ContactAndDateInfoIdUtil;
import com.oracle.calllog.analyze.util.DBTool;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;

public class MyDBRecordWriter extends RecordWriter<ContactDimBean, CountDuration> {
    //获取外键ID的工具类
    private ContactAndDateInfoIdUtil idUtil;
    //数据库的连接对象
    private Connection conn = null;
    //数据库SQL执行工具对象
    private DBTool dbTool;
    //编辑预处理对象
    private PreparedStatement preparedStatement = null;
    //SQL语句：用于将一条统计分析的结果记录增加到数据库中
    private String insertSQL = null;
    //一个批次待执行的的SQL总量计数
    private int count = 0;

    public MyDBRecordWriter() {
        this.conn = ConnectionPool.getInstance().getConnection();
    	dbTool = new DBTool();
        idUtil = new ContactAndDateInfoIdUtil(dbTool);
    }

    @Override
    public void write(ContactDimBean key, CountDuration value) throws IOException, InterruptedException {
        try {
            //tb_call
            //id_date_contact, id_date_dimension, id_contact, call_sum, call_duration_sum

            //时间维度表主键，表中包含year month day
            int dateId = idUtil.getDimensionId(key);
            //联系人表主键，表中包含telephone name
            int contactId = idUtil.getContactId(key);

            // tb_call主键，表中包含时间维度和联系人外键（外键约束在Phoenix中无效）
            // tb_call表的结构：id_date_contact, id_date_dimension, id_contact, call_sum, call_duration_sum
            String comboId = dateId + "_" + contactId;
            //通话总次数
            int callSum = Integer.valueOf(value.getCallSum());
            //通话总时长
            int callDurationSum = Integer.valueOf(value.getCallDurationSum());

            //向数据库中插入分析结果数据（Phoenix与RDBMS不同，分别处理）
            if (Constants.DB_TYPE.equals("phoenix")){
                if (insertSQL == null) {
                    //Phoenix天生就支持有则改，无则增的处理
                    insertSQL = "UPSERT INTO tb_call "
                            + "(id_date_contact, "
                            + "	id_date_dimension, "
                            + "	id_contact, "
                            + "	call_sum, "
                            + "	call_duration_sum) "
                            + "VALUES (?, ?, ?, ?, ?) ";
                }
            }else {
                if (insertSQL == null) {
                    //RDBMS中通过主键判断数据存在就修改，数据不存在就新增的语句用法
                    // 使用示例可参考ReadMe.txt
                    insertSQL = "INSERT INTO tb_call "
                            + "(id_date_contact, "
                            + "	id_date_dimension, "
                            + "	id_contact, "
                            + "	call_sum, "
                            + "	call_duration_sum) "
                            + "VALUES (?, ?, ?, ?, ?) "
                            + "ON DUPLICATE KEY UPDATE "
                            + "	id_date_contact = ?,"
                            + "	id_date_dimension = ?,"
                            + "	id_contact = ?,"
                            + "	call_sum = ?,"
                            + "	call_duration_sum = ?";
                }

            }
            //这样用性能不好，还是单独写一个能批处理的过程替换掉吧
            //dbTool.updateByPrepared(insertSQL, comboId,dateId,contactId,callSum,callDurationSum,comboId);

            if(preparedStatement == null){
                preparedStatement = dbTool.getPstmt(conn, insertSQL);
            }

            //本次SQL
            int i = 0;
            preparedStatement.setString(++i, comboId);
            preparedStatement.setInt(++i, dateId);
            preparedStatement.setInt(++i, contactId);
            preparedStatement.setString(++i, String.valueOf(callSum));
            preparedStatement.setString(++i, String.valueOf(callDurationSum));
            // 判断如果不是phoenix数据库，则为预编译的SQL语句再补充几个参数
            if(!Constants.DB_TYPE.equals("phoenix")){
                //无则插入，有则更新的判断依据
                preparedStatement.setString(++i, comboId);
                preparedStatement.setInt(++i, dateId);
                preparedStatement.setInt(++i, contactId);
                preparedStatement.setInt(++i, callSum);
                preparedStatement.setInt(++i, callDurationSum);
            }

            //将一条记录添加到批处理的批次中
            preparedStatement.addBatch();

            //累计一条未执行的SQL
            this.count++;
            //如果累计未执行的SQL达到了预设值，则批量执行一次SQL，并重新开始累计
            if(count >= Integer.parseInt(Constants.BATCH_MAXSIZE)){
                this.preparedStatement.executeBatch();
                this.conn.commit();
                this.count = 0;
                preparedStatement.clearBatch();
            }
        } catch (SQLException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void close(TaskAttemptContext context) throws IOException, InterruptedException {
        try {
            if(preparedStatement != null){
                preparedStatement.executeBatch();
                this.conn.commit();
            }
        } catch (SQLException e) {
            e.printStackTrace();
        }finally {
        	ConnectionPool.getInstance().revokeConnection(preparedStatement, conn);
        }
    }
}
